package com.guanzi.big_data_mq_process.consumer;

import com.guanzi.big_data_mq_process.constant.MessageConstant;
import com.guanzi.big_data_mq_process.model.OrderCreateContext;
import com.guanzi.big_data_mq_process.model.entity.TransactionMessage;
import com.guanzi.big_data_mq_process.model.enums.LocalTransactionStateEnum;
import com.guanzi.big_data_mq_process.model.enums.MessageStatusEnum;
import com.guanzi.big_data_mq_process.service.IOrdersService;
import com.guanzi.big_data_mq_process.service.ITransactionMessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.List;

import static com.guanzi.big_data_mq_process.model.enums.LocalTransactionStateEnum.PREPARED;
import static com.guanzi.big_data_mq_process.model.enums.LocalTransactionStateEnum.ROLLBACKED;

/**
 * 支付 消费者
 */
@Slf4j
@Service
@RocketMQMessageListener(topic = MessageConstant.ORDER_TOPIC,
        consumerGroup = MessageConstant.ORDER_CONSUMER_GROUP
//        selectorExpression = MessageConstant.ORDER_TAG,
//        selectorType = SelectorType.TAG
)
public class PaymentConsumer implements RocketMQListener<OrderCreateContext> {

    @Autowired
    private IOrdersService ordersService;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private ITransactionMessageService transactionMessageService;

    @Override
    public void onMessage(OrderCreateContext context) {
        String messageId = context.getMessageId();
        log.info("消费到 message: 【{}】的消息!", messageId);
        // 本地消息保存的是：事务半消息是否发送成功，所以如果消费者收到的是半消息，直接不处理
        LocalTransactionStateEnum state = transactionMessageService.selectTxState(messageId);
        if (List.of(ROLLBACKED, PREPARED).contains(state)) {
            return;
        }

        try {
            // 模拟支付订单，80%成功率
            boolean success = Math.random() < 0.8;
            log.info("message: 【{}】 消息是否支付成功: 【{}】", messageId, success);
            if (success) {
                // 更新订单状态(略) ...
                // 标记消息完成
                transactionMessageService.updateMsgStatus(messageId, MessageStatusEnum.COMPLETE);
            } else {
                handlePaymentFailure(context);
            }
        } catch (Exception e) {
            handlePaymentFailure(context);
        }
    }

    /**
     * 处理支付失败
     *  设置重试次数
     *  发送延迟消息
     * @param context
     */
    private void handlePaymentFailure(OrderCreateContext context) {
        TransactionMessage localMsg = transactionMessageService.getById(context.getMessageId());
        log.info("message: 【{}】 进入失败处理!", localMsg.getMessageId());
        if (localMsg.getRetryCount() >= 3) {
            transactionMessageService.updateMsgStatus(context.getMessageId(), MessageStatusEnum.WAIT_SENT);
        } else {
            // 发送延迟消息
            rocketMQTemplate.syncSendDelayTimeSeconds(
                    MessageConstant.ORDER_TOPIC + ":RETRY",
                    MessageBuilder.withPayload(context)
                            .setHeader(RocketMQHeaders.TRANSACTION_ID, context.getMessageId())
                            .build(),
                    300 // 5分钟
            );
            // 更新重试信息
            localMsg.setRetryCount(localMsg.getRetryCount() + 1);
            localMsg.setNextRetryTime(Timestamp.valueOf(LocalDateTime.now().plusMinutes(5)));
            transactionMessageService.updateById(localMsg);
        }
    }

}
